package pro.coman404.tools.table;

import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import java.time.LocalDateTime;

/**
 * @author NotFound404 at 2024/7/25
 * code code always code
 */
public class TableDemo3 {

    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        DataStream<Row> userStream = env.fromData(
                Row.of(LocalDateTime.parse("2021-08-21T13:00:00"),1,"Alice"),
                Row.of(LocalDateTime.parse("2021-08-21T13:05:00"),2,"Bob"),
                Row.of(LocalDateTime.parse("2021-08-21T13:10:00"),2,"Bob")
        ).returns(Types.ROW_NAMED(new String[]{"ts","uid","name"},Types.LOCAL_DATE_TIME,Types.INT,Types.STRING));

        DataStream<Row> orderStream = env.fromData(
                Row.of(LocalDateTime.parse("2021-08-21T13:02:00"),1,122),
                Row.of(LocalDateTime.parse("2021-08-21T13:07:00"),2,239),
                Row.of(LocalDateTime.parse("2021-08-21T13:11:00"),2,999)
        ).returns(Types.ROW_NAMED(new String[] {"ts","uid","amount"},Types.LOCAL_DATE_TIME,Types.INT,Types.INT));

        tableEnv.createTemporaryView("UserTable",userStream, Schema.newBuilder()
                        .column("ts", DataTypes.TIMESTAMP(3))
                        .column("uid", DataTypes.INT())
                        .column("name", DataTypes.STRING())
                        .watermark("ts","ts - INTERVAL '1' SECOND")
                .build());

        tableEnv.createTemporaryView("OrderTable",orderStream,Schema.newBuilder()
                .column("ts", DataTypes.TIMESTAMP(3))
                .column("uid", DataTypes.INT())
                .column("amount", DataTypes.INT())
                .watermark("ts","ts - INTERVAL '1' SECOND")
                .build());

        Table joinTable = tableEnv.sqlQuery("SELECT u.name,o.amount from UserTable u,OrderTable o " +
                "where u.uid=o.uid and o.ts BETWEEN u.ts AND u.ts + INTERVAL '5' MINUTES");

        DataStream<Row> joinStream = tableEnv.toDataStream(joinTable);

        joinStream.print();

        joinStream.keyBy(r->r.<String>getFieldAs("name"))
                .process(new KeyedProcessFunction<String, Row, String>() {

                    ValueState<String> seen;

                    @Override
                    public void open(OpenContext openContext) throws Exception {
                        seen = getRuntimeContext().getState(new ValueStateDescriptor<>("seen", TypeInformation.of(String.class)));
                    }

                    @Override
                    public void processElement(Row value, KeyedProcessFunction<String, Row, String>.Context ctx, Collector<String> out) throws Exception {
                        String name = value.<String>getFieldAs("name");
                        if (seen.value() == null){
                            seen.update(name);
                            out.collect(name);
                        }
                    }
                }).print();

        env.execute();

    }

}
